package com.atguigu.sparkcore.rdd.tranformdoublevalue

import com.atguigu.sparkcore.util.MySparkContextUtil
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

/**
  * 拉链
  * author 剧情再美终是戏
  * mail 13286520398@163.com
  * date 2020/1/6 17:49 
  * version 1.0
  **/
object Zip {

  def main(args: Array[String]): Unit = {

    // 获取sparkContext
    val sc = MySparkContextUtil.get(args)

    // 创建rdd
    val rdd: RDD[Int] = sc.makeRDD((1 to 6), 2)
    val rdd2: RDD[Int] = sc.makeRDD(Array(20, 30, 50), 2)

    // 操作rdd

    // １、spark zip
    // 要求：　分区数一样，分区内的个数一样
    //    val result = rdd.zip(rdd2)

    // 2、spark zipindexs
    //    val result = rdd.zipWithIndex()

    // 3、spark zippartition
    val result = rdd.zipPartitions(rdd2)((it1, it2) => {
      // scala zip (没有匹配的就扔掉)
      // 要求：分区数一致就可以了
      //      it1.zip(it2)

      // sacla zipall(没有匹配的就用默认值来补全位置)
      // 要求：分区数一致就可以了
      it1.zipAll(it2, -1, -2)
    })


    result.foreach(println)

    // 关闭资源
    MySparkContextUtil.close(sc)
  }

}
